package org.example.mq.consumer;

import com.alibaba.nacos.common.utils.MapUtils;
import com.aliyun.openservices.ons.api.ExpressionType;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.PropertyValueConst;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.common.message.MessageType;
import com.google.common.collect.Maps;
import java.util.Map;
import java.util.Properties;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.example.mq.MqConfig;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

/**
 * @title:
 * @description: 配置消费队列
 * @author: wuwenxue
 * @date: 2023-07-13 10:11
 */
@Component
@Slf4j
public class MQConsumerClient {

    @Resource
    private ApplicationContext wac;
    @Resource
    private MqConfig           mqConfig;

    @PostConstruct
    public void initConsumer() {
        String disabled = wac.getEnvironment().getProperty("mq.consumer.disabled", "false");
        if (StringUtils.equals(disabled, "true")) {
            return;
        }
        String appVersion = wac.getEnvironment().getProperty("app.version", "");
        if (StringUtils.isNoneBlank(appVersion)) {
            mqConfig.setGid(mqConfig.getGid().concat("_").concat(appVersion));
        }
        Map<String, MessageListenerConsumer> beans = wac.getBeansOfType(
                MessageListenerConsumer.class);
        if (beans.isEmpty()) {
            return;
        }
        //订阅集群消息
        Map<Subscription, MessageListener> subscriptionTable = Maps.newHashMap();
        //订阅广播消息
        Map<Subscription, MessageListener> broadcastSubscriptionTable = Maps.newHashMap();

        beans.values().forEach(messageListenerConsumer -> {
            messageListenerConsumer.setGid(mqConfig.getGid());
            Subscription sub = new Subscription();
            sub.setTopic(messageListenerConsumer.getTopic());
            sub.setExpression(buildExpression(messageListenerConsumer.getTag(), appVersion));
            sub.setType(ExpressionType.SQL92.name());
            log.info("订阅 : {}",sub);
            if (StringUtils.equals(messageListenerConsumer.getMessageModel(),
                    PropertyValueConst.CLUSTERING)) {
                subscriptionTable.put(sub, messageListenerConsumer);
            } else {
                broadcastSubscriptionTable.put(sub, messageListenerConsumer);
            }
            log.info("已订阅 messageModel = {}, Topic = {}, tag = {}",
                    messageListenerConsumer.getMessageModel(), messageListenerConsumer.getTopic(),
                    messageListenerConsumer.getTag());
        });

        if (MapUtils.isNotEmpty(subscriptionTable) && StringUtils.isBlank(mqConfig.getGid())) {
            throw new RuntimeException("缺失集群消息Gid配置");
        }

        if (MapUtils.isNotEmpty(broadcastSubscriptionTable) && StringUtils.isBlank(
                mqConfig.getGidBroadcast())) {
            throw new RuntimeException("缺失广播消息Gid配置");
        }
        //集群消息
        if (MapUtils.isNotEmpty(subscriptionTable)) {
            start(mqConfig.getGid(), PropertyValueConst.CLUSTERING, subscriptionTable);
        }
        //广播消息
        if (MapUtils.isNotEmpty(broadcastSubscriptionTable)) {
            start(mqConfig.getGidBroadcast(), PropertyValueConst.BROADCASTING,
                    broadcastSubscriptionTable);
        }
    }

    private String buildExpression(String tag, String appVersion) {
        StringBuilder expression = new StringBuilder();
        if (StringUtils.isNotEmpty(appVersion)) {
            expression.append("routeVersion = '").append(appVersion).append("'");
        }else {
            expression.append("routeVersion is null ");
        }
        if (StringUtils.isNotEmpty(tag) && StringUtils.equals("*", tag)) {
            expression.append(" and tags = '*' ");
        } else if (StringUtils.isNotEmpty(tag) && !StringUtils.equals("*", tag)) {
            String sTag = StringUtils.joinWith(",", StringUtils.split(tag, "||"));
            expression.append(" and tags is not null and tags in (").append(sTag).append(")");
        }
        return expression.toString();
    }

    private void start(String gid, String messageModel,
            Map<Subscription, MessageListener> subscriptionTable) {
        log.info("准备启动MQ消费者，消费者ID = {}", gid);
        ConsumerBean consumerBean = new ConsumerBean();
        Properties properties = new Properties();
        //您在控制台创建的Group ID
        properties.setProperty(PropertyKeyConst.GROUP_ID, gid);
        // 设置 TCP 接入域名(此处以公共云生产环境为例)，设置 TCP 接入域名，进入 MQ 控制台的消费者管理页面，在左侧操作栏单击获取接入点获取
        properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, mqConfig.getAddress());
        // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.setProperty(PropertyKeyConst.AccessKey, mqConfig.getAccessKey());
        // SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.setProperty(PropertyKeyConst.SecretKey, mqConfig.getSecretKey());
        // 消息模式
        properties.put(PropertyKeyConst.MessageModel, messageModel);
        //设置发送超时时间，单位毫秒
        properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis,
                mqConfig.getSendMsgTimeoutMillis());
        // 线程数
        properties.setProperty(PropertyKeyConst.ConsumeThreadNums, mqConfig.getConsumeThreadNums());

        consumerBean.setProperties(properties);

        consumerBean.setSubscriptionTable(subscriptionTable);

        consumerBean.start();
        log.info("MQ消费者已启动，消费者ID = {} , messageModel = {}", gid, messageModel);
    }

    @PreDestroy
    public void destroy() {
        log.info("销毁消费者");
    }
}
